package com.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Spark_Stream_WordCount {

  def main(args: Array[String]): Unit = {
    //创建环境对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming")
    //初始化 SparkStreamingContext             采集周期3秒
    val ssc = new StreamingContext(sparkConf,Seconds(3))

    //获取端口数据
    val lines:DStream[String] = ssc.socketTextStream("localhost",9999)
    //将每一行数据切分，形成一个个单词
    val words:DStream[String] = lines.flatMap(_.split(" "))
    //将单词 转换（映射）元组 (spark,1) (spark,1) (spark,1)
    val wordOne =  words.map( (_,1) )
    //(spark,1) (spark,1) (spark,1) ===> (spark,3)
    val wordCount = wordOne.reduceByKey(_+_)
    //打印
    wordCount.print()

    //启动SparkStreamingContext
    ssc.start()
    //等待采集器关闭
    ssc.awaitTermination()

  }

}
